package com.chukun.flink.stream.window.join;

import com.chukun.flink.stream.window.join.bean.ClickBean;
import com.chukun.flink.stream.window.join.bean.PrepareData;
import com.chukun.flink.stream.window.join.bean.UserTrade;
import com.chukun.flink.stream.window.join.extractor.ClickBeanEventTimestampExtractor;
import com.chukun.flink.stream.window.join.extractor.TradeEventTimestampExtractor;
import com.chukun.flink.stream.window.join.watermarker.ClickBeanWatermarkerGenerator;
import com.chukun.flink.stream.window.join.watermarker.TradeWatermarkerGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author chukun
 * @version 1.0.0
 * @description join基本操作， JoinFunction  FlatJoinFunction基本操作
 * @createTime 2022年05月26日 23:49:00
 */
public class TumblingWindowJoinOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<ClickBean> clickStream = env.fromCollection(PrepareData.getClicksData());

        DataStreamSource<UserTrade> tradeStream = env.fromCollection(PrepareData.getTradeData());

        // 设置原始用户点击数据的水印与事件时间
        SingleOutputStreamOperator<ClickBean> eventTimeClickStream = clickStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new ClickBeanWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new ClickBeanEventTimestampExtractor())
        );

        // 设置原始用户交易数据的水印与事件时间
        SingleOutputStreamOperator<UserTrade> eventTimeTradeStream = tradeStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new TradeWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new TradeEventTimestampExtractor())
        );

        JoinedStreams.WithWindow<ClickBean, UserTrade, String, TimeWindow> joinedWindow = eventTimeClickStream
                .join(eventTimeTradeStream)
                // 根据eventTimeClickStream的用户名对数据流元素分组
                .where(ClickBean::getUser)
                // 根据eventTimeTradeStream的用户名对数据流元素分组
                .equalTo(UserTrade::getUserName)
                // 基于事件时间的滚动窗口，为1小时
                .window(TumblingEventTimeWindows.of(Time.hours(1)));

        DataStream<String> applyJoinedStream = joinedWindow.apply(new JoinFunction<ClickBean, UserTrade, String>() {
            @Override
            public String join(ClickBean clickBean, UserTrade trade) throws Exception {
                return clickBean.toString() + "--->" + trade.toString();
            }
        });

        applyJoinedStream.print("JoinFunction输出");

        DataStream<String> applyFlatJoinStream = joinedWindow.apply(new FlatJoinFunction<ClickBean, UserTrade, String>() {
            @Override
            public void join(ClickBean clickBean, UserTrade userTrade, Collector<String> collector) throws Exception {
                // 将用户的访问时间和订单时间输出
                collector.collect(clickBean.getUser() + " : " + clickBean.getVisitTime() + " : " + userTrade.getTradeTime());
                // 将用户访问的url和终端类型合并输出
                collector.collect(clickBean.getUser() + ":" + clickBean.getUrl() + " : " + userTrade.getClient());
            }
        });

        applyFlatJoinStream.print("FlatJoinFunction输出");

        env.execute("TumblingWindowJoinOperator");
    }
}
